package com.exceeddata.vcloud.common.kafka.producer;

import com.exceeddata.vcloud.common.kafka.KafkaCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

/**
 * @author: yxh
 * @Description: Kafka生产者
 * @Date: Created in 2025/2/28 14:28
 * @Modified By:
 */
@Service
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 发送字符串消息到指定主题
     *
     * @param topic   主题名称
     * @param message 消息内容
     */
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }

    /**
     * 发送带键的字符串消息到指定主题
     *
     * @param topic   主题名称
     * @param key     消息键
     * @param message 消息内容
     */
    public void sendMessageWithKey(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message);
    }

    /**
     * 异步发送消息并处理回调
     *
     * @param topic   主题名称
     * @param message 消息内容
     * @param callback 回调函数
     */
    public void sendAsyncMessage(String topic, String message, KafkaCallback<String, String> callback) {
        kafkaTemplate.send(topic, message).addCallback(callback);
    }

    /**
     * 异步发送带键的消息并处理回调
     *
     * @param topic   主题名称
     * @param key     消息键
     * @param message 消息内容
     * @param callback 回调函数
     */
    public void sendAsyncMessageWithKey(String topic, String key, String message, KafkaCallback<String, String> callback) {
        kafkaTemplate.send(topic, key, message).addCallback(callback);
    }
}
